package main

import (
	"DMSKafka/config"
	"context"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"log"
	"sync"
	"time"
)

// 真实Topic
var realTopic = "realque"

// 延时Topic
var delayTopic = "testlog"
var wg sync.WaitGroup

func main() {
	conf, err := config.GetConfig()
	if err != nil {
		log.Printf("%+v\n", err)
		return
	}
	//初始化延迟消费者组
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = false
	group, err := sarama.NewConsumerGroup(conf.Host, "delay_group", config)
	if err != nil {
		log.Printf("%+v\n", errors.Wrap(err, "初始化生产者失败"))
		return
	}
	defer group.Close()
	//初始化延时发送的生产者
	proconfig := sarama.NewConfig()
	proconfig.Producer.Return.Successes = true
	proconfig.Producer.RequiredAcks = sarama.WaitForAll
	producer, err := sarama.NewSyncProducer(conf.Host, proconfig)
	if err != nil {
		log.Printf("%+v\n", errors.Wrap(err, "初始化生产者失败"))
		return
	}
	defer producer.Close()
	wg.Add(1)
	consumer := InitConsumer(producer, time.Second*10)
	go func() {
		for {
			if err = group.Consume(context.Background(), []string{delayTopic}, consumer); err != nil {
				break
			}
		}
		defer wg.Done()
	}()
	wg.Wait()

}

// 自定义实现接口
type Consumer struct {
	producer  sarama.SyncProducer
	delaytime time.Duration
}

func InitConsumer(producer sarama.SyncProducer, delaytime time.Duration) *Consumer {
	return &Consumer{producer: producer, delaytime: delaytime}
}

func (c *Consumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		log.Printf("DelayMessage: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
		//获取当前时间
		now := time.Now()
		//判段是否超时，超时的消息发送到真实队列
		if now.Sub(message.Timestamp) >= c.delaytime {
			msg := &sarama.ProducerMessage{
				Topic: realTopic,
				Key:   sarama.ByteEncoder(message.Key),
				Value: sarama.ByteEncoder(message.Value),
			}
			_, _, err := c.producer.SendMessage(msg)
			if err == nil {
				session.MarkMessage(message, "")
				continue
			}
			log.Printf("%+v\n", errors.Wrap(err, "发送消息失败"))
			return nil
		}
		time.Sleep(time.Second)
		return nil
	}
	return nil
}
